#include <iterator>
#include <unistd.h>
#include "MarkupSTL.h"
#include "global.h"
#include "cmd.h"
#include "Game.h"
#include "RedisClient.h"
#include "Logic.h"
#include "json.h"
#include "globalExtern.h"
#include "Host.h"
#include "1000_flow.pb.h"
#include "1700_account.pb.h"


#define  MOD_UID (100)
char* CGameServer::s_szexchangetype = "direct";
char* CGameServer::s_szexchange = "flow";
char* CGameServer::s_szexAgencyChange = "agency_account";
//char* CGameServer::s_szexchange1 = "flow_event";

CGameServer::CGameServer():m_mqPort(0), m_mqconn(NULL), m_mqchannel(1), m_mqchannel1(2)
{
}

CGameServer::~CGameServer()
{
// 	if (m_mqconn1)
// 		releasemq1();
	if (m_mqconn)
		releasemq();

}

bool CGameServer::InitGame()
{
	char szconfig[] = "./config.xml";
	if (!ReadCommonConf(szconfig))
		return false;

	return initmq();
// 	if (!initmq())
// 		return false;
// 
// 	return initmq1();
}

bool CGameServer::initmq(void)
{
	m_mqconn = amqp_new_connection();
	if (!m_mqconn)
	{
		log_error("create mq conn err");
		return false;
	}
	amqp_socket_t* socket = amqp_tcp_socket_new(m_mqconn);
	if (!socket)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("create mq TCP socket err");
		return false;
	}
	int status = amqp_socket_open(socket, m_mqIp.c_str(), m_mqPort);
	if (status)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq socket err");
		return false;
	}
	amqp_rpc_reply_t x = amqp_login(m_mqconn, m_mqVhost.c_str(), 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, m_mqUsername.c_str(), m_mqPwd.c_str());
	if (AMQP_RESPONSE_NORMAL != x.reply_type)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("mq login fail,host:%s,user:%s,pwd:%s", m_mqVhost.c_str(), m_mqUsername.c_str(), m_mqPwd.c_str());
		return false;
	}
	amqp_channel_open_ok_t* t = amqp_channel_open(m_mqconn, m_mqchannel);
	if (!t)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq channel fail");
		return false;
	}
	// 	if (0 == t->channel_id.len || !t->channel_id.bytes)
	// 	{
	// 		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
	// 		amqp_destroy_connection(m_mqconn);
	// 		m_mqconn = NULL;
	// 		log_error("open mq channel fail");
	// 		return false;
	// 	}
	// 	log_debug("mq channle:%s", std::string((char *)t->channel_id.bytes, t->channel_id.len).c_str());

	amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x1.reply_type)
	{
		releasemq();
		log_error("mq open channel amqp_get_rpc_reply fail");
		return false;
	}

	amqp_exchange_declare(m_mqconn, m_mqchannel, amqp_cstring_bytes(CGameServer::s_szexchange), amqp_cstring_bytes(s_szexchangetype), false, true, false, false, amqp_empty_table);

	amqp_rpc_reply_t x2 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x2.reply_type)
	{
		releasemq();
		log_error("mq open amqp_exchange_declare fail");
		return false;
	}

	amqp_exchange_declare(m_mqconn, m_mqchannel, amqp_cstring_bytes(CGameServer::s_szexAgencyChange), amqp_cstring_bytes(s_szexchangetype), false, true, false, false, amqp_empty_table);

	amqp_rpc_reply_t x3 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x3.reply_type)
	{
		releasemq();
		log_error("mq open amqp_exchange_declare1 fail");
		return false;
	}

	return true;
}

// bool CGameServer::initmq1(void)
// {
// 	m_mqconn1 = amqp_new_connection();
// 	if (!m_mqconn1)
// 	{
// 		log_error("create mq conn err");
// 		return false;
// 	}
// 	amqp_socket_t* socket = amqp_tcp_socket_new(m_mqconn1);
// 	if (!socket)
// 	{
// 		amqp_destroy_connection(m_mqconn1);
// 		m_mqconn1 = NULL;
// 		log_error("create mq TCP socket err");
// 		return false;
// 	}
// 	int status = amqp_socket_open(socket, m_mqIp.c_str(), m_mqPort);
// 	if (status)
// 	{
// 		amqp_destroy_connection(m_mqconn1);
// 		m_mqconn1 = NULL;
// 		log_error("open mq socket err");
// 		return false;
// 	}
// 	amqp_rpc_reply_t x = amqp_login(m_mqconn1, m_mqVhost.c_str(), 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, m_mqUsername.c_str(), m_mqPwd.c_str());
// 	if (AMQP_RESPONSE_NORMAL != x.reply_type)
// 	{
// 		amqp_connection_close(m_mqconn1, AMQP_REPLY_SUCCESS);
// 		amqp_destroy_connection(m_mqconn1);
// 		m_mqconn1 = NULL;
// 		log_error("mq login fail,host:%s,user:%s,pwd:%s", m_mqVhost.c_str(), m_mqUsername.c_str(), m_mqPwd.c_str());
// 		return false;
// 	}
// 	amqp_channel_open_ok_t* t = amqp_channel_open(m_mqconn1, m_mqchannel1);
// 	if (!t)
// 	{
// 		amqp_connection_close(m_mqconn1, AMQP_REPLY_SUCCESS);
// 		amqp_destroy_connection(m_mqconn1);
// 		m_mqconn1 = NULL;
// 		log_error("open mq channel fail");
// 		return false;
// 	}
// 
// 	amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn1);
// 	if (AMQP_RESPONSE_NORMAL != x1.reply_type)
// 	{
// 		releasemq1();
// 		log_error("mq open channel amqp_get_rpc_reply fail");
// 		return false;
// 	}
// 
// 	amqp_exchange_declare(m_mqconn1, m_mqchannel1, amqp_cstring_bytes(CGameServer::s_szexchange1), amqp_cstring_bytes(s_szexchangetype), false, true, false, false, amqp_empty_table);
// 
// 	amqp_rpc_reply_t x2 = amqp_get_rpc_reply(m_mqconn1);
// 	if (AMQP_RESPONSE_NORMAL != x2.reply_type)
// 	{
// 		releasemq1();
// 		log_error("mq open amqp_exchange_declare fail");
// 		return false;
// 	}
// 
// 	return true;
// }

void CGameServer::releasemq(void)
{
	if (m_mqconn)
	{
		amqp_channel_close(m_mqconn, m_mqchannel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
	}

	return;
}

// void CGameServer::releasemq1(void)
// {
// 	if (m_mqconn1)
// 	{
// 		amqp_channel_close(m_mqconn1, m_mqchannel1, AMQP_REPLY_SUCCESS);
// 		amqp_connection_close(m_mqconn1, AMQP_REPLY_SUCCESS);
// 		amqp_destroy_connection(m_mqconn1);
// 		m_mqconn1 = NULL;
// 	}
// 
// 	return;
// }

int CGameServer::OnBYTimeOut(int Timerid)
{
	return 0;
}


int CGameServer::ProcessPacket(SocketHandler * pHandler, NETInputPacket* inputPkg)
{
	unsigned int cmd = inputPkg->GetCmdType();
	switch (cmd)
	{
	case flow::FLOW_CMD_FLOW_PUSH:
	{
		procFlowPush(pHandler, inputPkg);
		break;
	}
	case account_proto::ACCOUNT_CMD_REGISTER_NEW_PLAYER_PUSH:
	{
		procAgencyAccountMQ(pHandler, inputPkg);
		break;
	}
	default:
		break;
	}

	return 0;
}

int CGameServer::procFlowPush(SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	procFlowMQ(pHandler, inputPkg);
	return 0;
}

int CGameServer::procFlowMQ(SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	flow::FlowPush msg;
	if (!inputPkg->parsePb(&msg))
	{
		log_error("parse FlowPush fail ");
		return 0;
	}

	time_t curt = 0;
	unsigned int uid = 0;

	int count = msg.flow_size();
	
	for (int i = 0; i < count; ++i)
	{
		uid = msg.flow(i).uid();
		if (0 == msg.flow(i).optime())
		{
			if (0 == curt)
				curt = time(NULL);

			flow::Flow* flow = msg.mutable_flow(i);
			flow->set_optime(curt);
			log_debug("set time,uid,%d,act,%d", msg.flow(i).uid(), msg.flow(i).op().act());
		}	

	}

	if (!m_mqconn)
	{
		if (initmq())
		{
			log_alert("mq reconnect OK");
		}
		else
		{
			log_error("mq reconnect fail");
		}
	}

	if (uid > 0)
	{
		string str;
		msg.SerializeToString(&str);
		log_debug("procAssetFlowMQ,uid:%u,data len:%u",uid, str.size());
		if (m_mqconn)
		{
			amqp_basic_properties_t props;
			props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
			props.delivery_mode = 2; /* persistent delivery mode */

			unsigned int index = uid % MOD_UID;
			amqp_bytes_t message;
			message.bytes = const_cast<char*>(str.data());
			message.len = str.size();
			int ret = amqp_basic_publish(m_mqconn, m_mqchannel, amqp_cstring_bytes(s_szexchange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
			log_debug("procAssetFlowMQ,uid:%u,data len:%u,ret:%x", uid, str.size(), ret);
			if (AMQP_STATUS_OK != ret)
			{
				log_error("amqp_basic_publish fail,ret:%d,msg len:%u", ret, message.len);
				if (AMQP_STATUS_CONNECTION_CLOSED == ret || AMQP_STATUS_SOCKET_ERROR == ret)
				{
					releasemq();
					if (initmq())
					{
						ret = amqp_basic_publish(m_mqconn, m_mqchannel, amqp_cstring_bytes(s_szexchange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
						log_alert("mq reconnect OK, publish ret:%d",ret);
					}
					else
					{
						log_error("reconnect fail,ret:%d", ret);
					}
				}
			}
		}
	}

	return 0;
}

int CGameServer::procAgencyAccountMQ(SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	account_proto::AccountInfo msg;
	if (!inputPkg->parsePb(&msg))
	{
		log_error("parse FlowPush fail ");
		return 0;
	}
	log_debug("procAgencyAccountMQ account=%s",msg.ShortDebugString().c_str());
	//m_pHost->SendtoSrv(account_proto::ACCOUNT_CMD_REGISTER_NEW_PLAYER_PUSH, SERVER_TYPE_STATISTIC, &msg, msg.uid());
	time_t curt = 0;
	unsigned int agency_id = msg.agency_id();

	if (!m_mqconn)
	{
		//if (initmq1())
		if (initmq())
		{
			log_alert("mq1 reconnect OK");
		}
		else
		{
			log_error("mq1 reconnect fail");
		}
	}

	if (agency_id > 0)
	{
		string str;
		msg.SerializeToString(&str);
		log_debug("procAgencyAccountMQ,agency_id:%u,data len:%u",agency_id, str.size());
		if (m_mqconn)
		{
			amqp_basic_properties_t props;
			props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
			props.delivery_mode = 2; /* persistent delivery mode */

			unsigned int index = agency_id % MOD_UID;
			amqp_bytes_t message;
			message.bytes = const_cast<char*>(str.data());
			message.len = str.size();
			//int ret = amqp_basic_publish(m_mqconn1, m_mqchannel1, amqp_cstring_bytes(s_szexAgencyChange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
			int ret = amqp_basic_publish(m_mqconn, m_mqchannel, amqp_cstring_bytes(s_szexAgencyChange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
			log_debug("procAssetFlowMQ,agency_id:%u,data len:%u,ret:%x", agency_id, str.size(), ret);
			if (AMQP_STATUS_OK != ret)
			{
				log_error("amqp_basic_publish fail,ret:%d,msg len:%u", ret, message.len);
				if (AMQP_STATUS_CONNECTION_CLOSED == ret || AMQP_STATUS_SOCKET_ERROR == ret)
				{
					releasemq();
					if (initmq())
					{
						//ret = amqp_basic_publish(m_mqconn1, m_mqchannel1, amqp_cstring_bytes(s_szexAgencyChange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
						ret = amqp_basic_publish(m_mqconn, m_mqchannel, amqp_cstring_bytes(s_szexAgencyChange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
						log_alert("mq reconnect OK, publish ret:%d",ret);
					}
					else
					{
						log_error("reconnect fail,ret:%d", ret);
					}
				}
			}
		}
	}

	return 0;
}

bool CGameServer::ReadCommonConf(const char* strxml)
{
	CMarkupSTL  markup;
	if (!markup.Load(strxml))
	{
		log_error("Load xml %s failed", strxml);
		return false;
	}

	if (!markup.FindElem("application"))
	{
		log_error("Not Found Elem [application] in xml[%s]", strxml);
		return false;
	}

	if (!markup.IntoElem())
	{
		log_error("IntoElem [application] failed.");
		return false;
	}


	if (!markup.FindElem("mq"))
	{
		log_error("Not Found Elem [mq] in xml[%s]", strxml);
		return false;
	}
	else
	{
		m_mqIp = markup.GetAttrib("ip");
		m_mqPort = atoi(markup.GetAttrib("port").c_str());
		m_mqVhost = markup.GetAttrib("vhost");
		m_mqUsername = markup.GetAttrib("username");
		m_mqPwd = markup.GetAttrib("password");
	}

	if (m_mqIp.empty() || m_mqVhost.empty() || m_mqUsername.empty() || m_mqPwd.empty() || 0 == m_mqPort)
		return false;

	log_debug("ReadCommonConf succ!!");

	return true;
}